package com.edu.realtime.app.dws;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.edu.realtime.app.func.DimAsyncFunction;
import com.edu.realtime.bean.TestPaperUcScoreDuration;
import com.edu.realtime.util.DateFormatUtil;
import com.edu.realtime.util.MyClickhouseUtil;
import com.edu.realtime.util.MyKafkaUtil;

/**
 * Created on 2022/10/19.
 *
 * @author Topus
 * @desc 考试域：考试人数、总分、总时长
 * @tip 需要启动DwdTestExam类
 */
public class DwsTestPaperUcScoreDurationWindow {
    public static void main(String[] args) throws Exception {
        /*
        使用状态编程做：
        1)dwd_test_exam读取数据
        2)维护状态变量总分数和总时长
         */
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_test_exam";
        String groupId = "dws_test_paper_uc_score_duration_window_group";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));
        SingleOutputStreamOperator<JSONObject> jsonObjDS = source.map(JSONObject::parseObject);
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = jsonObjDS
            .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000L;
                    }
                }));
        // map
        SingleOutputStreamOperator<TestPaperUcScoreDuration> beanDS = withWatermarkDS.map(jsonObj -> {
            String paperId = jsonObj.getString("paper_id");
            Double score = jsonObj.getDouble("score");
            Long durationSec = jsonObj.getLong("duration_sec");
            String userId = jsonObj.getString("user_id");
            return new TestPaperUcScoreDuration("", "", paperId, "", 1, durationSec, score, 0, 0L, userId);
        });

        // 分组
        KeyedStream<TestPaperUcScoreDuration, String> keyedDS = beanDS.keyBy(TestPaperUcScoreDuration::getPaperId);

        // 过滤做状态变成获得paper试卷粒度的独立用户
        SingleOutputStreamOperator<TestPaperUcScoreDuration> processDS =
            keyedDS.process(new KeyedProcessFunction<String, TestPaperUcScoreDuration, TestPaperUcScoreDuration>() {

                // 状态编程处理独立uv
                private MapState<String, Integer> mapState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    // 状态变成设置1day的有效期
                    MapStateDescriptor<String, Integer> stateDescriptor =
                        new MapStateDescriptor<>("test-paper-map-state", String.class, Integer.class);
                    stateDescriptor.enableTimeToLive(
                        StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.days(1)).build());
                    mapState = getRuntimeContext().getMapState(stateDescriptor);
                }

                @Override
                public void processElement(TestPaperUcScoreDuration paperUcScoreDuration,
                    KeyedProcessFunction<String, TestPaperUcScoreDuration, TestPaperUcScoreDuration>.Context ctx,
                    Collector<TestPaperUcScoreDuration> out) throws Exception {

                    String userId = paperUcScoreDuration.getUserId();
                    if (!mapState.contains(userId)) {
                        paperUcScoreDuration.setUserCount(1);
                        mapState.put(userId, 1);
                    }
                    out.collect(paperUcScoreDuration);
                }
            });

        // 开窗
        AllWindowedStream<TestPaperUcScoreDuration, TimeWindow> windowDS =
            processDS.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<TestPaperUcScoreDuration> reduceDS =
            windowDS.reduce(new ReduceFunction<TestPaperUcScoreDuration>() {
                @Override
                public TestPaperUcScoreDuration reduce(TestPaperUcScoreDuration value1, TestPaperUcScoreDuration value2)
                    throws Exception {
                    value1.setTestCount(value1.getTestCount() + value2.getTestCount());
                    value1.setTotalScore(value1.getTotalScore() + value2.getTotalScore());
                    value1.setTotalDuration(value1.getTotalDuration() + value2.getTotalDuration());
                    value1.setUserCount(value1.getUserCount() + value2.getUserCount());
                    return value1;
                }
            }, new AllWindowFunction<TestPaperUcScoreDuration, TestPaperUcScoreDuration, TimeWindow>() {
                @Override
                public void apply(TimeWindow window, Iterable<TestPaperUcScoreDuration> values,
                    Collector<TestPaperUcScoreDuration> out) throws Exception {
                    String sst = DateFormatUtil.toYmdHms(window.getStart());
                    String edt = DateFormatUtil.toYmdHms(window.getEnd());
                    for (TestPaperUcScoreDuration examStatsBean : values) {
                        examStatsBean.setStt(sst);
                        examStatsBean.setEdt(edt);
                        examStatsBean.setTs(System.currentTimeMillis());
                        out.collect(examStatsBean);
                    }
                }
            });

        reduceDS.print("reduce");
        // 聚合后关联获取试卷信息
        // 写DimAsyncFunction处理异步请求，使用模板方法实现处理从Phoenix查询主键维度数据
        SingleOutputStreamOperator<TestPaperUcScoreDuration> joinDS =
            AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TestPaperUcScoreDuration>("dim_test_paper") {
                @Override
                public void join(TestPaperUcScoreDuration input, JSONObject dimInfo) {
                    input.setPaperTitle(dimInfo.getString("PAPER_TITLE"));
                }

                @Override
                public String getKey(TestPaperUcScoreDuration obj) {
                    return obj.getPaperId();
                }
            }, 120, TimeUnit.SECONDS);

        // 写入clickhouse
        joinDS.addSink(MyClickhouseUtil.<TestPaperUcScoreDuration>getSinkFunction(
            "insert into dws_test_paper_uc_score_duration_window values(?,?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}
